package com.zy.gmall.realtime.util;

import com.alibaba.fastjson.JSONObject;
import jdk.internal.org.objectweb.asm.Type;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import scala.reflect.internal.Types;

import java.io.IOException;
import java.util.List;


public class HbaseUtil {
    //获取Hbase连接
    public static Connection getConnect(){
        try {
            Configuration conf = new Configuration();
            // 配置 Zookeeper
            conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
            Connection connection = ConnectionFactory.createConnection(conf);
            return connection;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    //获取支持异步操作的连接对象
    public static AsyncConnection getAsyncConnection(){
        Configuration conf = new Configuration();
        // 配置 Zookeeper
        conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
        try {
            return ConnectionFactory.createAsyncConnection(conf).get();
        } catch (Exception e) {
            throw new RuntimeException();
        }
    }

    public static void closeAsyncConnection(AsyncConnection asyncConnection){
        if (asyncConnection != null || !asyncConnection.isClosed()){
            try {
                asyncConnection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

    }

    /**
     * 以异步的方式从hbase维度表中查询维度数据
     * @param conn          异步连接客户端
     * @param nameSpace     表空间名
     * @param tableName     表名
     * @param rowkey        主键
     * @return      将查询出来的维度数据封装为一个json对象
     */
    public static JSONObject getDimInfoFromHbaseByAsync(AsyncConnection conn,String nameSpace,String tableName,String rowkey){
        try {
            TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
            AsyncTable<AdvancedScanResultConsumer> asyncTable = conn.getTable(tableNameObj);
            Get get = new Get(Bytes.toBytes(rowkey));
            Result result = asyncTable.get(get).get();
            List<Cell> cells = result.listCells();
            if(cells != null && cells.size() > 0){
                JSONObject jsonObj = new JSONObject();
                for (Cell cell : cells) {
                    String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                    jsonObj.put(columnName,columnValue);
                }
                return jsonObj;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }


    //关闭连接
    public static void closeHbase(Connection connection){
        if (connection != null || !connection.isClosed()){
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

    }






    //创建
    public static void createHbaseTable(Connection conf,String namespace,String table,String ... lists){
        //判断列组是否为空
        if (lists.length < 1 || lists == null){
            throw new RuntimeException("列族不存在");
        }
        try(Admin admin = conf.getAdmin()) {
            TableName tableName = TableName.valueOf(namespace, table);
            if (admin.tableExists(tableName)){
                System.out.println("要创建的" + namespace + ":" + tableName + "已存在");
                return;
            }
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(
                    tableName
            );

            //加入列族
            for (String list : lists) {
                ColumnFamilyDescriptor columnFamilyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(list)).build();
                tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor);
            }

            //创建
            System.out.println("创建" + namespace + ":" + tableName);
            admin.createTable(tableDescriptorBuilder.build()

            );
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    //删除
    public static void deleteHbaseTable(Connection conf, String namespace,String table){
        //admin 表操作
        try (Admin admin = conf.getAdmin()){
            TableName tableName = TableName.valueOf(namespace,table);
            //判空
            if (admin.tableExists(tableName)){
                System.out.println("不存在");
                return;
            }
            //删除
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    //删除一条数据
    public static void delData(Connection conn, String namespace,String table,String rowKey) throws IOException {
        TableName tableName = TableName.valueOf(namespace,table);
        Table connTable = conn.getTable(tableName);
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        connTable.delete(delete);
        connTable.close();
    }

    //新增数据
    public static void addData(Connection conn, String namespace,String rowKey,String table,String family,String[] lists,String[] values) throws IOException {
        TableName tableName = TableName.valueOf(namespace,table);
        Table connTable = conn.getTable(tableName);
        Put put = new Put(Bytes.toBytes(rowKey));

        for (int i = 0; i < lists.length; i++) {
            String columnName = lists[i];
            String columnValue = values[i];
            if (columnValue != null){
                put.addColumn(Bytes.toBytes(family),Bytes.toBytes(columnName),Bytes.toBytes(columnValue));
            }
        }
        connTable.put(put);
        connTable.close();
    }
    public static String getDimDicHbaseTable(){
        return "CREATE TABLE dimDicTable (\n" +
                " dic_code STRING,\n" +
                " info ROW<dic_name STRING> ,\n" +
                " PRIMARY KEY (dic_code) NOT ENFORCED\n" +
                ")" + getHbaseConnector("gmall_realtime:dim_base_dic");
    }
    public static String getHbaseConnector(String table_name){
        return "WITH (\n" +
                " 'connector' = 'hbase-2.2',\n" +
                " 'table-name' = '"+ table_name +"',\n" +
                " 'zookeeper.quorum' = 'hadoop102:2181,hadoop103:2181,hadoop104:2181',\n" +
                " 'lookup.async' = 'true',\n" +
                " 'lookup.cache' = 'PARTIAL',\n" +
                " 'lookup.partial-cache.max-rows' = '500',\n" +
                " 'lookup.partial-cache.expire-after-write' = '1 hour',\n" +
                " 'lookup.partial-cache.expire-after-access' = '1 hour'\n" +
                ")";
    }

    public static JSONObject getDimInfoFromHbase(Connection conn,String nameSpace,String tableName,String rowKey){
        TableName tableNameObj = TableName.valueOf(nameSpace, tableName);//获取表名
        //获取表
        try(Table table = conn.getTable(tableNameObj)) {
            //根据
            Get get = new Get(Bytes.toBytes(rowKey));
            //得到一行结果
            Result result = table.get(get);
            //转换成list
            List<Cell> cells = result.listCells();
            if (cells != null && cells.size() > 0){
                JSONObject jsonObj = new JSONObject();
                for (Cell cell : cells) {
                    String name = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    jsonObj.put(name,value);
                }
                return jsonObj;
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
}
